package com.qingcloud.df.executor.cluster;

import cn.hutool.core.map.MapUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.qingcloud.df.sdk.component.config.ExecutorConfig;
import com.qingcloud.df.sdk.component.context.CompInstContext;
import com.qingcloud.df.sdk.component.context.ConfigManager;
import com.qingcloud.df.sdk.component.enums.EnvEnum;
import com.qingcloud.df.sdk.component.enums.NodeTypeEnum;
import com.qingcloud.df.sdk.component.message.NodeInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.ThreadLocalRandom;
import java.util.stream.Collectors;

public class ClusterManager {

    private static NodeInfo currentNode;
    /**
     * 节点信息
     */
    private static final Map<String, NodeInfo> NODES_LIST = Maps.newConcurrentMap();
    /**
     * 存储所有实例信息，按节点编号，JOB编号分组
     */
    private static final Map<String, Map<Integer,List<CompInstContext>>> NODE_JOB_INSTANCD_LIST = Maps.newConcurrentMap();


    /**
     * 同步刷新节点
     * @param nodes
     */
    public static synchronized void refreshNodeList(Map<String, NodeInfo> nodes) {
        NODES_LIST.clear();
        NODES_LIST.putAll(nodes);
    }

    /**
     * 新增Node
     * @param nodeId
     * @param nodeInfo
     */
    public static synchronized void registerNode(String nodeId, NodeInfo nodeInfo) {
        NODES_LIST.put(nodeId,nodeInfo);
    }

    /**
     * 删除节点
     * @param nodeId
     */
    public static synchronized void removeNode(String nodeId) {
        //删除节点上的所有job
        removeNodeJobs(nodeId);
        //删除节点
        NODES_LIST.remove(nodeId);
    }

    /**
     * 同步刷新所有节点所有Job分配信息  nodeId,<jobId,instanceList>
     * @param compInst
     */
    public static synchronized void refreshJobInstance(Map<String, Map<Integer,List<CompInstContext>>> compInst) {
        NODE_JOB_INSTANCD_LIST.clear();
        NODE_JOB_INSTANCD_LIST.putAll(compInst);
    }

    public static synchronized void pubNode(String nodeId, Map<Integer,List<CompInstContext>> jobMap) {
        NODE_JOB_INSTANCD_LIST.put(nodeId,jobMap);
    }

    /**
     * 新增一个job
     * @param nodeId
     * @param jobId
     * @param compInstContexts
     */
    public static synchronized void pubNodeJob(String nodeId, Integer jobId,List<CompInstContext> compInstContexts) {
        Map<Integer,List<CompInstContext>> jobMap  = NODE_JOB_INSTANCD_LIST.get(nodeId);
        if(jobMap!=null && jobMap.size()>0){
            jobMap.put(jobId,compInstContexts);
        }else {
            jobMap = Maps.newConcurrentMap();
            jobMap.put(jobId,compInstContexts);
            NODE_JOB_INSTANCD_LIST.put(nodeId,jobMap);
        }
    }

    /**
     * 增加一个实例
     * @param nodeId
     * @param jobId
     * @param compInstContext
     */
    public static void pubNodeInstance(String nodeId,Integer jobId, CompInstContext compInstContext) {
        Map<Integer, List<CompInstContext>> jobList =  NODE_JOB_INSTANCD_LIST.get(nodeId);
        if(jobList == null || jobList.size()==0){
            jobList =  Maps.newConcurrentMap();
            List<CompInstContext> list =  new ArrayList<>();
            list.add(compInstContext);
            jobList.put(jobId,list);
            NODE_JOB_INSTANCD_LIST.put(nodeId,jobList);
        }else{
            List<CompInstContext> compInstContexts = jobList.get(jobId);
            if(compInstContexts==null || compInstContexts.size()==0){
                compInstContexts  = new ArrayList<>();
            }
            //删除已经存在的实例
            compInstContexts.removeIf( x->x.getId().intValue()==compInstContext.getId());
            //添加新实例
            compInstContexts.add(compInstContext);
            jobList.put(jobId,compInstContexts);
        }
        NODE_JOB_INSTANCD_LIST.put(nodeId,jobList);
    }

    /**
     * 清除一个job
     * @param nodeId
     * @param jobId
     */
    public static synchronized void removeNodeJob(String nodeId, Integer jobId) {
       if(NODE_JOB_INSTANCD_LIST.get(nodeId)!=null && NODE_JOB_INSTANCD_LIST.size()>0){
           NODE_JOB_INSTANCD_LIST.get(nodeId).remove(jobId);
       }
    }

    /**
     * 清除一个实例
     * @param nodeId
     * @param jobId
     * @param instanceId
     */
    public static synchronized void removeNodeInstance(String nodeId, Integer jobId,Integer instanceId) {

        List<CompInstContext> contexts = getInstanceByJobId(nodeId,jobId);
        if(contexts!=null) {
            contexts.removeIf( x->x.getId().intValue()==instanceId);
        }
    }

    /**
     * 删除一个节点上的所有job信息
     * @param nodeId
     */
    public static synchronized void removeNodeJobs(String nodeId) {
        NODE_JOB_INSTANCD_LIST.remove(nodeId);
    }




    public static Map<String, Map<Integer,List<CompInstContext>>> get() {
        return NODE_JOB_INSTANCD_LIST;
    }

    /**
     * 获取一个一个已分配作业的所有实例信息
     * @param jobId
     * @return
     */
    public static List<CompInstContext> getInstanceByJobId(Integer jobId) {
        List<CompInstContext>  compInstContexts = new ArrayList<>();
        NODE_JOB_INSTANCD_LIST.forEach((key,map)->{
            map.forEach((job,data)->{
                if(job.intValue() == jobId){
                    compInstContexts.addAll(data);
                }
            });
        });
        return compInstContexts;
    }

    public static CompInstContext getContextByInstanceId(Integer jobId,Integer instanceId) {
        for(Map.Entry<String, Map<Integer,List<CompInstContext>>> enrty: NODE_JOB_INSTANCD_LIST.entrySet()){
            Map<Integer,List<CompInstContext>> map = enrty.getValue();
            if(map!=null) {
                List<CompInstContext> compInstContexts = map.get(jobId);
                if (compInstContexts != null && compInstContexts.size() > 0) {
                    for (CompInstContext context : compInstContexts) {
                        if (context.getId().intValue() == instanceId) {
                            return context;
                        }
                    }
                }
            }
        }
        return null;
    }

    /**
     * 获取指定节点的指定job的所有实例
     * @param nodeId
     * @param jobId
     * @return
     */
    public static List<CompInstContext> getInstanceByJobId(String nodeId ,Integer jobId) {
        Map<Integer,List<CompInstContext>> jobMap =  NODE_JOB_INSTANCD_LIST.get(nodeId);
        if(jobMap!=null && jobMap.size()>0){
            return jobMap.get(jobId);
        }
        return null;
    }
    /**
     * 判断一个job是否已经分配给其他节点 (节点未分配给当前指定节点）
     * @param jobId
     * @param nodeId 当前节点
     * @return
     */
    public static boolean hasAssigned(Integer jobId,String nodeId) {
        if(NODE_JOB_INSTANCD_LIST!=null && NODE_JOB_INSTANCD_LIST.size()>0){
            for(Map.Entry<String,  Map<Integer, List<CompInstContext>>> entry : NODE_JOB_INSTANCD_LIST.entrySet()){
                String key  = entry.getKey();
                Map<Integer, List<CompInstContext>> map = entry.getValue();
                if(map.containsKey(jobId)&& !key.equals(nodeId)){
                    return true;
                }
            }
        }
        return false;
    }

    /**
     * 节点已分配
     * @param jobId
     * @return
     */
    public static boolean hasAssigned(Integer jobId) {
        if(NODE_JOB_INSTANCD_LIST!=null && NODE_JOB_INSTANCD_LIST.size()>0){
            for(Map.Entry<String,  Map<Integer, List<CompInstContext>>> entry : NODE_JOB_INSTANCD_LIST.entrySet()){
                Map<Integer, List<CompInstContext>> map = entry.getValue();
                if(map.containsKey(jobId)){
                    return true;
                }
            }
        }
        return false;
    }


    /**
     * 查找Job分配的节点，未分配返回空
     * @param jobId
     * @return
     */
    public static String getNodeByJobId(Integer jobId) {
        String nodeId =  null;
        if(NODE_JOB_INSTANCD_LIST!=null && NODE_JOB_INSTANCD_LIST.size()>0){
            for(Map.Entry<String,  Map<Integer, List<CompInstContext>>> entry : NODE_JOB_INSTANCD_LIST.entrySet()){
                Map<Integer, List<CompInstContext>> map = entry.getValue();
                nodeId = entry.getKey();
                if(map.containsKey(jobId) && map.get(jobId)!=null && map.get(jobId).size()>0){
                    return nodeId;
                }
            }
        }
        return nodeId;
    }


    /**
     * 获取instanceId指定Node 地址
     * @param instanceId
     * @return
     */
    public static String getNodeByInstanceId(Integer jobId, Integer instanceId) {

        if(NODE_JOB_INSTANCD_LIST!=null && NODE_JOB_INSTANCD_LIST.size()>0){
            for(Map.Entry<String,  Map<Integer, List<CompInstContext>>> entry : NODE_JOB_INSTANCD_LIST.entrySet()){
                Map<Integer, List<CompInstContext>> map = entry.getValue();
                String nodeId= entry.getKey();
                if(jobId!=null) {
                    List<CompInstContext> contexts = map.get(jobId);
                    if (findNode(instanceId, contexts)) {
                        return nodeId;
                    }
                }else{
                    for(Map.Entry<Integer, List<CompInstContext>> compInstMap : map.entrySet()){
                        List<CompInstContext> contexts = compInstMap.getValue();
                        if (findNode(instanceId,contexts)){
                            return nodeId;
                        }
                    }
                }
            }
        }
        return null;
    }

    private static boolean findNode(Integer instanceId, List<CompInstContext> contexts) {
        if (contexts != null && contexts.size() > 0) {
            for (CompInstContext context : contexts) {
                if (context.getId().intValue() == instanceId) {
                    return true;
                }
            }
        }
        return false;
    }


    /**
     * 获取节点中所有作业实例
     * @param nodeId
     * @return
     */
    public static Map<Integer,List<CompInstContext>> getJobByNodeId(String nodeId) {
        return Optional.ofNullable(NODE_JOB_INSTANCD_LIST.get(nodeId))
                .orElseGet(HashMap::new);
    }


    public static Map<String, NodeInfo> getNodeMap() {
        return NODES_LIST;
    }

    public static NodeInfo getNodeById(String nodeId) {
        return NODES_LIST.get(nodeId);
    }
    public static synchronized void setCurrentNode(NodeInfo currentNode) {
        ClusterManager.currentNode = currentNode;
    }
    public static NodeInfo getCurrentNode() {
        if(currentNode==null) {
            currentNode = new NodeInfo();
            currentNode.setNodeType(ExecutorConfig.getType());
            currentNode.setAkkaAddress(ExecutorConfig.getAkkaNodeAddress());
            currentNode.setEnv(ExecutorConfig.getEnv());
            currentNode.setStatus(1);
            currentNode.setLastHeartbeatTime(System.currentTimeMillis());
        }
        return currentNode;
    }

    /**
     * 加载本地所有配置文件信息
     */
    public static  void init(){
        Logger logger = LoggerFactory.getLogger(ClusterManager.class);
        ConfigManager.reloadByEnv(EnvEnum.DEV.name);
        ConfigManager.reloadByEnv(EnvEnum.PRO.name);

//        ConcurrentMap<Integer, CompInstContext> devConfig  = ConfigManager.getDevConfig();
//        ConcurrentMap<Integer, CompInstContext>  prdConfig  = ConfigManager.getProConfig();
//
//        ConcurrentMap<Integer, List<CompInstContext>>  devJobMap = Maps.newConcurrentMap();
//        ConcurrentMap<Integer, List<CompInstContext>>  prdJobMap = Maps.newConcurrentMap();
//        if(devConfig!=null) {
////            logger.debug("dev instance config size {}!",devConfig.size());
////            List<CompInstContext> contexts  = null;
////            for (Map.Entry<Integer, CompInstContext> entry : devConfig.entrySet()) {
////                CompInstContext compInstContext = entry.getValue();
////                if(devJobMap.get(compInstContext.getJobId())==null){
////                    contexts  = new ArrayList<>();
////                    contexts.add(compInstContext);
////                    devJobMap.put(compInstContext.getJobId(),contexts);
////                }else {
////                    devJobMap.get(compInstContext.getJobId()).add(compInstContext);
////                }
////            }
////            logger.debug("have {} dev job will startup !",devJobMap.size());
////        }
//        if(prdConfig!=null) {
//            logger.debug("prod instance config size {}!",prdConfig.size());
//            List<CompInstContext> contexts  = null;
//            for (Map.Entry<Integer, CompInstContext> entry : prdConfig.entrySet()){
//                CompInstContext compInstContext = entry.getValue();
//                if(prdJobMap.get(compInstContext.getJobId())==null){
//                    contexts  = new ArrayList<>();
//                    contexts.add(compInstContext);
//                    prdJobMap.put(compInstContext.getJobId(),contexts);
//                }else {
//                    prdJobMap.get(compInstContext.getJobId()).add(compInstContext);
//                }
//            }
//            logger.debug("have {} prod job will startup !",prdJobMap.size());
//        }
//        ConfigManager.refreshJobInstance(devJobMap);
    }


    public static NodeInfo selectAvailableNode(String env) {
        Map<String, NodeInfo> nodeInfoMap = ClusterManager.getNodeMap();
        if (MapUtil.isEmpty(nodeInfoMap)) {
            return null;
        }
        List<NodeInfo> nodeList = nodeInfoMap.entrySet().stream()
                .map(entry -> entry.getValue())
                .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(nodeList)) {
            return null;
        }
        // 获取可用的NODE.
        List<NodeInfo> availableNode = new ArrayList<>();
        for(NodeInfo nodeinfo : nodeList){
            if((nodeinfo.getNodeType().equals(NodeTypeEnum.EXECUTOR.name)
                    ||nodeinfo.getNodeType().indexOf(NodeTypeEnum.EXECUTOR.name)>-1)
                && nodeinfo.getEnv().equals(env)
                && nodeinfo.getStatus()==1
                && !nodeinfo.isOverload()){
                availableNode.add(nodeinfo);
            }
        }

        if (CollectionUtils.isEmpty(availableNode)) {
            return null;
        }
        int index = ThreadLocalRandom.current().nextInt(availableNode.size());
        return availableNode.get(index);
    }
//
//    public List<WorkerInfo> getSuitableWorkers(JobInfoDO jobInfo) {
//
//        List<WorkerInfo> workers = Lists.newLinkedList(getWorkerInfosByAppId(jobInfo.getAppId()).values());
//
//        workers.removeIf(workerInfo -> filterWorker(workerInfo, jobInfo));
//
//        DispatchStrategy dispatchStrategy = DispatchStrategy.of(jobInfo.getDispatchStrategy());
//        switch (dispatchStrategy) {
//            case RANDOM:
//                Collections.shuffle(workers);
//                break;
//            case HEALTH_FIRST:
//                workers.sort((o1, o2) -> o2.getSystemMetrics().calculateScore() - o1.getSystemMetrics().calculateScore());
//                break;
//            default:
//                // do nothing
//        }
//
//        // 限定集群大小（0代表不限制）
//        if (!workers.isEmpty() && jobInfo.getMaxWorkerCount() > 0 && workers.size() > jobInfo.getMaxWorkerCount()) {
//            workers = workers.subList(0, jobInfo.getMaxWorkerCount());
//        }
//        return workers;
//    }
//
//    public List<WorkerInfo> getAllWorkers(Long appId) {
//        List<WorkerInfo> workers = Lists.newLinkedList(getWorkerInfosByAppId(appId).values());
//        workers.sort((o1, o2) -> o2.getSystemMetrics().calculateScore() - o1.getSystemMetrics().calculateScore());
//        return workers;
//    }
//
//    /**
//     * get all alive workers
//     *
//     * @param appId appId
//     * @return alive workers
//     */
//    @DesignateServer
//    public List<WorkerInfo> getAllAliveWorkers(Long appId) {
//        List<WorkerInfo> workers = Lists.newLinkedList(getWorkerInfosByAppId(appId).values());
//        workers.removeIf(WorkerInfo::timeout);
//        return workers;
//    }
//
//    /**
//     * Gets worker info by address.
//     *
//     * @param appId   the app id
//     * @param address the address
//     * @return the worker info by address
//     */
//    public Optional<WorkerInfo> getWorkerInfoByAddress(Long appId, String address) {
//        final Map<String, WorkerInfo> workerInfosByAppId = getWorkerInfosByAppId(appId);
//        if (null != workerInfosByAppId && null != address) {
//            return Optional.ofNullable(workerInfosByAppId.get(address));
//        }
//        return Optional.empty();
//    }
}
